package main;

import helper.CompositeKey;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class PairMapper extends Mapper<Object, Text, CompositeKey, IntWritable>{

    private CompositeKey outKey = new CompositeKey();
    private IntWritable outValue = new IntWritable(1);
    
    private int user1 = 0;
    private int user2 = 0;
    
    public void map(Object key, Text value, Context context) 
            throws IOException, InterruptedException {
        
        String followers_str = value.toString().split(":")[1];
        String[] followers = followers_str.split(" ");
        int length = followers.length;
        
        for(int i=0; i<length; i++) {
            for(int j=i+1; j<length; j++) {
                user1 = Integer.parseInt(followers[i]);
                user2 = Integer.parseInt(followers[j]);
                
                outKey.setUser(user1);
                outKey.setSimilar(user2);
                context.write(outKey, outValue);
//                System.out.println(pair.toString());
                
                outKey.setUser(user2);
                outKey.setSimilar(user1);
                context.write(outKey, outValue);
//                System.out.println(pair.toString());
            }
        }
        
    }
    
}